package com.ds.lens.flink.monitor.datasource.function;

import com.ds.lens.data.common.document.datasource.LensDsSchema;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.types.Row;

import java.sql.Timestamp;
import java.text.ParseException;
import java.text.SimpleDateFormat;

/**
 * @author WeiShaoying
 * @date 2020/4/28 上午11:26
 */
@Slf4j
public class DataSourceWatermarks extends AscendingTimestampExtractor<Row> {

    private LensDsSchema schema;
    private static SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

    public DataSourceWatermarks(LensDsSchema schema){
        this.schema = schema;
    }


    @Override
    public long extractAscendingTimestamp(Row row) {
        String value = row.getField(schema.getIndex()).toString();
        switch (schema.getType()){
            case "Long":
                return Long.valueOf(value);
            case "String":
                try {
                    return simpleDateFormat.parse(value).getTime();
                } catch (ParseException e) {
                    log.error("parse String type date {} failed ", value);
                    return 0;
                }
            case "Timestamp":
                return Timestamp.valueOf(value).getTime();
        }
        return 0;
    }
}
